package com.bw.yk06;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;
import java.util.TimeZone;

public class test10 {
    public static void main(String[] args) throws Exception {
        //创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);
        //读取kafka
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "test3");
        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<>("tms_ods", new SimpleStringSchema(), properties));

        SingleOutputStreamOperator<String> process = stream
                .filter(x -> JSON.parseObject(x).getString("table").equals("order_info") &&
                        JSON.parseObject(x).getJSONObject("data").getString("status").equals("60050"))
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        return JSON.parseObject(s).getJSONObject("data");
                    }
                }).assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((event, timestamp) -> {
                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                            sdf.setTimeZone(TimeZone.getTimeZone("GMT+8"));
                            try {
                                return sdf.parse(event.getString("create_time")).getTime();
                            } catch (ParseException e) {
                                throw new RuntimeException(e);
                            }
                        }))
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10),Time.seconds(-5)))
                .process(new ProcessAllWindowFunction<JSONObject, String, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<JSONObject, String, TimeWindow>.Context context, Iterable<JSONObject> iterable, Collector<String> collector) throws Exception {
                        Double amount = 0.0;
                        Integer cnts = 0;
                        for (JSONObject jsonObject : iterable) {
                            cnts++;
                            amount = amount + jsonObject.getDouble("amount");
                        }
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String stt = sdf.format(context.window().getStart());
                        String edt = sdf.format(context.window().getEnd());
                        collector.collect(stt + "," + edt + "," + amount + "," + cnts);
                    }
                });

        process.print();


        env.execute();
    }
}
